package xmq

import (
	"context"
	"fmt"
	"gitee.com/go-mid/infra/xmq/mqconf"
	"gitee.com/go-mid/infra/xmq/xdelay"
	"gitee.com/go-mid/infra/xmq/xkafka"
	"gitee.com/go-mid/infra/xmq/xpulsar"
	"gitee.com/go-mid/infra/xobserver"
	"sync"

	"gitee.com/go-mid/infra/xconfig"
	"gitee.com/go-mid/infra/xlog"
)

type InstanceConf struct {
	//配置中心的namespace
	Namespace string
	//具体的某个配置key
	Key string
}

func (c *InstanceConf) String() string {
	return getInstanceKey(c.Namespace, c.Key)
}
func getInstanceKey(namespace, key string) string {
	return fmt.Sprintf("%s-%s", namespace, key)
}

type InstanceManager struct {
	//配置中心
	configCenter xconfig.ConfigCenter

	instances sync.Map
	mutex     sync.Mutex
}

func NewInstanceManager(configCenter xconfig.ConfigCenter) *InstanceManager {
	im := &InstanceManager{configCenter: configCenter}
	return im
}

func (m *InstanceManager) buildKey(conf *InstanceConf) string {
	return conf.String()
}

func (m *InstanceManager) add(conf *InstanceConf, in interface{}) {
	m.instances.Store(conf.String(), in)
}

func (m *InstanceManager) newInstance(ctx context.Context, conf *InstanceConf) (interface{}, error) {

	var cConf mqconf.MQConfig
	if err := m.configCenter.UnmarshalKey(ctx, conf.Namespace, conf.Key, &cConf); err != nil {
		return nil, err
	}

	switch cConf.MQType {
	case mqconf.MQType_Kafka_Reader:
		var rConf xkafka.ReaderConfig
		if err := m.configCenter.UnmarshalKey(ctx, conf.Namespace, conf.Key, &rConf); err != nil {
			return nil, err
		}
		cli := xkafka.NewKafkaReader(&rConf)
		return cli, nil
	case mqconf.MQType_Kafka_Writer:
		var rConf xkafka.WriterConfig
		if err := m.configCenter.UnmarshalKey(ctx, conf.Namespace, conf.Key, &rConf); err != nil {
			return nil, err
		}
		cli := xkafka.NewKafkaWriter(&rConf)
		return cli, nil
	case mqconf.MQType_Pulsar_Reader:
		var rConf xpulsar.ConsumerConfig
		if err := m.configCenter.UnmarshalKey(ctx, conf.Namespace, conf.Key, &rConf); err != nil {
			return nil, err
		}
		return xpulsar.NewConsumer(ctx, &rConf)
	case mqconf.MQType_Pulsar_Writer:
		var rConf xpulsar.ProducerConfig
		if err := m.configCenter.UnmarshalKey(ctx, conf.Namespace, conf.Key, &rConf); err != nil {
			return nil, err
		}
		return xpulsar.NewProducer(ctx, &rConf)
	case mqconf.MQType_Delay_Reader:
		var rConf xdelay.DelayMqConfig
		if err := m.configCenter.UnmarshalKey(ctx, conf.Namespace, conf.Key, &rConf); err != nil {
			return nil, err
		}
		ins := xdelay.NewDelayClient(rConf)
		return ins, nil
	case mqconf.MQType_Delay_Writer:
		var rConf xdelay.DelayMqConfig
		if err := m.configCenter.UnmarshalKey(ctx, conf.Namespace, conf.Key, &rConf); err != nil {
			return nil, err
		}
		ins := xdelay.NewDelayClient(rConf)
		return ins, nil
	default:
		return nil, fmt.Errorf("cache type %s error", cConf.MQType)
	}
}

func (m *InstanceManager) get(ctx context.Context, conf *InstanceConf) interface{} {
	fun := "InstanceManager.get -->"

	var err error
	var in interface{}
	key := m.buildKey(conf)
	in, ok := m.instances.Load(key)
	if ok == false {

		m.mutex.Lock()

		in, ok = m.instances.Load(key)
		if ok {
			m.mutex.Unlock()
			return in
		}

		xlog.Infof(ctx, "%s newInstance, conf: %v", fun, conf)
		in, err = m.newInstance(ctx, conf)
		if err != nil {
			xlog.Errorf(ctx, "%s NewInstance err, conf: %v, err: %s", fun, conf, err.Error())
			m.mutex.Unlock()
			return nil
		}
		m.configCenter.RegisterObserver(ctx, conf.Namespace, xobserver.NewConfigObserver(m.applyChangeEvent))
		in, _ = m.instances.LoadOrStore(key, in)

		m.mutex.Unlock()
	}
	return in
}

func (m *InstanceManager) applyChange(ctx context.Context, namespace, k string, change *xobserver.Change) {
	fun := "InstanceManager.applyChange-->"
	xlog.Infof(ctx, "%s apply change:%v to key:%s", fun, change, k)
	insKey := getInstanceKey(namespace, k)
	if old, ok := m.instances.Load(insKey); ok && old != nil {
		if err := m.closeInstance(ctx, old); err != nil {
			xlog.Errorf(ctx, "%s close instance err:namespace: %s, key: %s, err: %v", fun, namespace, k, err)
		}
		m.instances.Delete(insKey)
	}
	if in, err := m.newInstance(ctx, &InstanceConf{
		Namespace: namespace,
		Key:       k,
	}); err == nil && in != nil {
		m.instances.Store(insKey, in)
		return
	}
	xlog.Errorf(ctx, "change instance error: namespace: %s, k: %s", namespace, k)
	return
}

func (m *InstanceManager) applyChangeEvent(ctx context.Context, ce *xobserver.ChangeEvent) {
	xlog.Infof(ctx, "got new change event:%v", ce)
	for key, change := range ce.Changes {
		// NOTE: 只需关心 MODIFY 与 DELETE 类型改变
		if change.ChangeType != xobserver.MODIFY && change.ChangeType != xobserver.DELETE {
			continue
		}

		m.applyChange(ctx, ce.Namespace, key, change)
	}
}

func (m *InstanceManager) Close() {
	fun := "InstanceManager.Close -->"

	ctx := context.TODO()

	m.instances.Range(func(key, value interface{}) bool {
		xlog.Infof(ctx, "%s key:%v", fun, key)
		err := m.closeInstance(ctx, value)
		if err != nil {
			xlog.Errorf(ctx, "%s close instance err:%v", fun, err)
		}
		m.instances.Delete(key)
		return true
	})
}

func (m *InstanceManager) closeInstance(ctx context.Context, instance interface{}) error {
	fun := "InstanceManager.closeInstance-->"
	if ins, ok := instance.(MQCloser); ok && ins != nil {
		ins.Close(ctx)
		return nil
	}
	xlog.Warnf(ctx, "%s instantce not close might memory leak", fun)
	return nil
}
